home *** CD-ROM | disk | FTP | other *** search
/ Mac Easy 2010 May / Mac Life Ubuntu.iso / casper / filesystem.squashfs / usr / lib / python2.6 / dist-packages / checkbox / contrib / REThread.pyc (.txt) < prev   
Encoding:
Python Compiled Bytecode  |  2009-10-12  |  5.9 KB  |  143 lines
  1. # Source Generated with Decompyle++
  2. # File: in.pyc (Python 2.6)
  3.  
  4. '''Enhanced threading.Thread which can deliver a return value and propagate
  5. exceptions from the called thread to the calling thread.
  6.  
  7. Copyright (C) 2007 Canonical Ltd.
  8. Author: Martin Pitt <martin.pitt@ubuntu.com>
  9.  
  10. This program is free software; you can redistribute it and/or modify it
  11. under the terms of the GNU General Public License as published by the
  12. Free Software Foundation; either version 2 of the License, or (at your
  13. option) any later version.  See http://www.gnu.org/copyleft/gpl.html for
  14. the full text of the license.
  15. '''
  16. import threading
  17. import sys
  18.  
  19. class REThread(threading.Thread):
  20.     '''Enhanced threading.Thread which can deliver a return value and propagate
  21.     exceptions from the called thread to the calling thread.'''
  22.     
  23.     def __init__(self, group = None, target = None, name = None, args = (), kwargs = { }, verbose = None):
  24.         '''Initialize Thread, identical to threading.Thread.__init__().'''
  25.         threading.Thread.__init__(self, group, target, name, args, kwargs, verbose)
  26.         self._REThread__target = target
  27.         self._REThread__args = args
  28.         self._REThread__kwargs = kwargs
  29.         self._retval = None
  30.         self._exception = None
  31.  
  32.     
  33.     def run(self):
  34.         '''Run target function, identical to threading.Thread.run().'''
  35.         if self._REThread__target:
  36.             
  37.             try:
  38.                 self._retval = self._REThread__target(*self._REThread__args, **self._REThread__kwargs)
  39.             self._exception = sys.exc_info()
  40.  
  41.         
  42.  
  43.     
  44.     def return_value(self):
  45.         '''Return value from target function.
  46.  
  47.         This can only be called after the thread has finished, i. e. when
  48.         isAlive() is False and did not terminate with an exception.'''
  49.         if not not self.isAlive():
  50.             raise AssertionError
  51.         if not not (self._exception):
  52.             raise AssertionError
  53.         return self._retval
  54.  
  55.     
  56.     def exc_info(self):
  57.         '''Return a tuple (type, value, traceback) of the exception caught in
  58.         run().'''
  59.         return self._exception
  60.  
  61.     
  62.     def exc_raise(self):
  63.         '''Raises the exception caught in the thread.
  64.  
  65.         Does nothing if no exception was caught.'''
  66.         if self._exception:
  67.             raise self._exception[0], self._exception[1], self._exception[2]
  68.         self._exception
  69.  
  70.  
  71. if __name__ == '__main__':
  72.     import unittest
  73.     import time
  74.     import traceback
  75.     import exceptions
  76.     
  77.     def idle(seconds):
  78.         '''Test thread to just wait a bit.'''
  79.         time.sleep(seconds)
  80.  
  81.     
  82.     def div(x, y):
  83.         '''Test thread to divide two numbers.'''
  84.         return x / y
  85.  
  86.     
  87.     class _REThreadTest(unittest.TestCase):
  88.         
  89.         def test_return_value(self):
  90.             '''Test that return value works properly.'''
  91.             t = REThread(target = div, args = (42, 2))
  92.             t.start()
  93.             t.join()
  94.             t.exc_raise()
  95.             self.assertEqual(t.return_value(), 21)
  96.             self.assertEqual(t.exc_info(), None)
  97.  
  98.         
  99.         def test_no_return_value(self):
  100.             '''Test that REThread works if run() does not return anything.'''
  101.             t = REThread(target = idle, args = (0.5,))
  102.             t.start()
  103.             self.assertRaises(AssertionError, t.return_value)
  104.             t.join()
  105.             self.assertEqual(t.return_value(), None)
  106.             self.assertEqual(t.exc_info(), None)
  107.  
  108.         
  109.         def test_exception(self):
  110.             '''Test that exception in thread is caught and passed.'''
  111.             t = REThread(target = div, args = (1, 0))
  112.             t.start()
  113.             t.join()
  114.             self.assertRaises(AssertionError, t.return_value)
  115.             self.assert_(t.exc_info()[0] == exceptions.ZeroDivisionError)
  116.             exc = traceback.format_exception(t.exc_info()[0], t.exc_info()[1], t.exc_info()[2])
  117.             self.assert_(exc[-1].startswith('ZeroDivisionError'))
  118.             self.assert_(exc[-2].endswith('return x / y\n'))
  119.  
  120.         
  121.         def test_exc_raise(self):
  122.             '''Test that exc_raise() raises caught thread exception.'''
  123.             t = REThread(target = div, args = (1, 0))
  124.             t.start()
  125.             t.join()
  126.             self.assertRaises(AssertionError, t.return_value)
  127.             raised = False
  128.             
  129.             try:
  130.                 t.exc_raise()
  131.             except:
  132.                 raised = True
  133.                 e = sys.exc_info()
  134.                 exc = traceback.format_exception(e[0], e[1], e[2])
  135.                 self.assert_(exc[-1].startswith('ZeroDivisionError'))
  136.                 self.assert_(exc[-2].endswith('return x / y\n'))
  137.  
  138.             self.assert_(raised)
  139.  
  140.  
  141.     unittest.main()
  142.  
  143.